package com.lyx.rpc.consumer.common;

import com.lyx.rpc.annotation.RpcService;
import com.lyx.rpc.common.helper.RpcServiceHelper;
import com.lyx.rpc.common.threadpool.ClientThreadPool;
import com.lyx.rpc.consumer.common.handler.RpcConsumerHandler;
import com.lyx.rpc.consumer.common.helper.RpcConsumerHandlerHelper;
import com.lyx.rpc.consumer.common.initializer.RpcConsumerInitializer;
import com.lyx.rpc.protocol.RpcProtocol;
import com.lyx.rpc.protocol.meta.ServiceMeta;
import com.lyx.rpc.protocol.request.RpcRequest;
import com.lyx.rpc.proxy.api.consumer.Consumer;
import com.lyx.rpc.proxy.api.future.RPCFuture;
import com.lyx.rpc.registry.api.RegistryService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author by lyx
 * 服务消费者
 */
public class RpcConsumer implements Consumer {
    private final Logger logger = LoggerFactory.getLogger(RpcConsumer.class);
    private final Bootstrap bootstrap;
    private final EventLoopGroup eventLoopGroup;
    // 通过单例模式 双检索 确保在整个应用程序中只有一个RpcConsumer实例
    private static volatile RpcConsumer instance;
    // 用于缓存已建立的连接对应的RpcConsumerHandler。这样可以保证服务消费者与特定服务提供者之间的连接唯一性，避免重复建立连接。
    private static Map<String, RpcConsumerHandler> handlerMap = new ConcurrentHashMap<>();

    private RpcConsumer() {
        bootstrap = new Bootstrap();
        eventLoopGroup = new NioEventLoopGroup(4);
        bootstrap.group(eventLoopGroup)
            .channel(NioSocketChannel.class)
            .handler(new RpcConsumerInitializer());
    }

    public static RpcConsumer getInstance() {
        if (instance == null) {
            synchronized (RpcConsumer.class) {
                if (instance == null) {
                    instance = new RpcConsumer();
                }
            }
        }
        return instance;
    }

    /**
     * 为了保证服务消费者与特定服务提供者之间的连接的唯一性，
     * 会将服务消费者与特定服务提供者的连接保存到一个Map结构中。
     * 当某个服务消费者与特定的服务提供者再次进行数据交互时，
     * 只需要从Map中获取连接即可，不需要重复建立连接。
     *
     * @param protocol
     * @throws Exception
     */
    public RPCFuture sendRequest(RpcProtocol<RpcRequest> protocol, RegistryService registryService) throws Exception {
        RpcRequest request = protocol.getBody();
        String serviceKey = RpcServiceHelper.buildServiceKey(request.getClassName(), request.getVersion(), request.getGroup());
        Object[] parameters = request.getParameters();
        int invokerHashCode = (parameters == null || parameters.length <= 0) ? serviceKey.hashCode() : parameters[0].hashCode();
        ServiceMeta serviceMeta = registryService.discovery(serviceKey, invokerHashCode);
        if (serviceMeta != null) {
            RpcConsumerHandler rpcConsumerHandler = RpcConsumerHandlerHelper.get(serviceMeta);
            // 缓存中无rpcConsumerHandler
            if (rpcConsumerHandler == null) {
                rpcConsumerHandler = this.getRpcConsumerHandler(serviceMeta.getServiceAddr(), serviceMeta.getPort());
                RpcConsumerHandlerHelper.put(serviceMeta, rpcConsumerHandler);
            } else if (!rpcConsumerHandler.getChannel().isActive()) { // 缓存中存在RpcClientHandler，但不活跃
                rpcConsumerHandler.close();
                rpcConsumerHandler = this.getRpcConsumerHandler(serviceMeta.getServiceAddr(), serviceMeta.getPort());
                RpcConsumerHandlerHelper.put(serviceMeta, rpcConsumerHandler);
            }
            return rpcConsumerHandler.sendRequest(protocol, request.getAsync(), request.getOneway());
        }
        return null;
    }

    /**
     * 创建连接并返回RpcClientHandler
     */
    private RpcConsumerHandler getRpcConsumerHandler(String serviceAddress, int port) throws InterruptedException {
        ChannelFuture future = bootstrap.connect(serviceAddress, port).sync();
        future.addListener((ChannelFutureListener) channelFuture -> {
            if (channelFuture.isSuccess()) {
                logger.info("connect rpc server {} on port {} success.", serviceAddress, port);
            } else {
                logger.error("connect rpc server {} on port {} failed.", serviceAddress, port);
                channelFuture.cause().printStackTrace();
                eventLoopGroup.shutdownGracefully();
            }
        });
        return future.channel().pipeline().get(RpcConsumerHandler.class);
    }


    public void close() {
        RpcConsumerHandlerHelper.closeRpcConsumerHandler();
        eventLoopGroup.shutdownGracefully();
        ClientThreadPool.shutdown();
    }
}
